package cn.doitedu.dao;

import cn.doitedu.cache.CacheManager;
import cn.doitedu.pojo.CombineCondition;
import cn.doitedu.pojo.EventBean;
import cn.doitedu.pojo.EventParam;
import cn.doitedu.pojo.LogBean;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.java.utils.ParameterTool;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 从Doris中查询用户的行为数据
 * 特点：查询速度快，并发能力强，并且支持多表join
 */
public class HistoryEventDorisDaoImpl implements HistoryEventDao {

    private Connection connection;

    private CacheManager cacheManager;

    @Override
    public void init(ParameterTool parameterTool, MapState<String, String> cacheState) throws Exception {
        String url = parameterTool.getRequired("doris.url");
        String username = parameterTool.getRequired("doris.username");
        String password = parameterTool.getRequired("doris.password");

        connection = DriverManager.getConnection(url, username, password);

        //初始化cacheManager
        String cacheManagerImpl = parameterTool.getRequired("cache.manager.impl");
        cacheManager = (CacheManager) Class.forName(cacheManagerImpl).newInstance();
        cacheManager.init(parameterTool, cacheState);
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }

    @Override
    public String queryEventStr(LogBean bean, CombineCondition combineCondition) throws Exception {

        //检测是否有符合条件的缓存可用
        //取出条件
        String deviceId = bean.getDeviceId();
        String cacheId = combineCondition.getCacheId();
        Long startTime = combineCondition.getStartTime();
        Long endTime = combineCondition.getEndTime();

        //最外面的大Key
        String bigKey = deviceId;
        //里面的小Key
        String smallKey = cacheId+ ":" +startTime + ":" + endTime;

        //第一种情况：直接命中缓存
        //已缓存的数据：t4---t8
        //要查询的数据：t4---t8
        String seqStr = cacheManager.getSeqStr(bigKey, smallKey);
        if (seqStr != null) return seqStr;

        //第二种情况：部分命中缓存
        //已缓存的数据：t4---t8
        //要查询的数据：t4--------t10
        //可复用的数据：t4---t8，然后查询t9-t10的数据，再拼接到一起
        List<String> times = cacheManager.getTimeRange(bigKey);
        for (String cacheIdAndTime : times) {
            String[] fields = cacheIdAndTime.split(":");
            String c = fields[0];
            long s = Long.parseLong(fields[1]);
            long e = Long.parseLong(fields[2]);
            if (cacheId.equals(c) && startTime == s && endTime > e) {
                //从redis中获取缓存的数据
                String seqStr1 = cacheManager.getSeqStr(bigKey, c + ":" + startTime + ":" + e);
                //查询数据库返回其他部分的数据
                combineCondition.setStartTime(e);
                combineCondition.setEndTime(endTime);
                String seqStr2 = queryByTimeRange(deviceId, combineCondition);
                String res = seqStr1 + seqStr2;
                //将查询的数据设置到缓存中
                cacheManager.setSeqStr(bigKey, smallKey, res, 60 * 60 * 1000 * 12);
                return res;
            }
        }


        //第三中情况：部分命中缓存
        //已缓存的数据：    t4---t8
        //要查询的数据：t2-------t8
        //可复用的数据：t4---t8，然后查询t2-t4的数据，再拼接到一起
        for (String cacheIdAndTime : times) {
            String[] fields = cacheIdAndTime.split(":");
            String c = fields[0];
            long s = Long.parseLong(fields[1]);
            long e = Long.parseLong(fields[2]);
            if(cacheId.equals(c) && startTime < s && endTime == e) {
                String seqStr1 = cacheManager.getSeqStr(bigKey, cacheId + ":" + s + ":" + endTime);
                combineCondition.setStartTime(startTime);
                combineCondition.setEndTime(s);
                String seqStr2 = queryByTimeRange(deviceId, combineCondition);
                String res = seqStr2 + seqStr1;
                //将查询的数据设置到缓存中
                cacheManager.setSeqStr(bigKey, smallKey, res, 60 * 60 * 1000 * 12);
                return res;
            }
        }

        //前面的缓存都没有命中（必须查询数据库）
        String res = queryByTimeRange(deviceId, combineCondition);
        cacheManager.setSeqStr(bigKey, smallKey, res, 60 * 60 * 1000 * 12);
        return res;




    }


    private String queryByTimeRange(String deviceId, CombineCondition combineCondition) throws Exception {
        //查询ClickHouse，并且将行为字符串返回
        StringBuffer sb = new StringBuffer();
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        try {
            preparedStatement = connection.prepareStatement(combineCondition.getSql());
            preparedStatement.setString(1, deviceId);
            preparedStatement.setLong(2, combineCondition.getStartTime());
            preparedStatement.setLong(3, combineCondition.getEndTime());
            //返回的resultSet中，装的是多个行为
            resultSet = preparedStatement.executeQuery();
            //用来装实际从clickhouse中查询出的事件（各种各样的事件，可能很多个）
            ArrayList<String> events = new ArrayList<>();
            while (resultSet.next()) {
                String eventStr = resultSet.getString(1);
                events.add(eventStr);
            }
            //我实际关系的即要匹配的事件
            List<String> targetEvents = combineCondition.getEvents().stream().map(EventParam::getEventId).collect(Collectors.toList());
            //将实际查出来的事件与要匹配的事件完成映射
            for (String event : events) {
                int index = targetEvents.indexOf(event);
                sb.append(index+1);
            }
            //完成正则匹配的功能
        } finally {
            if (resultSet != null) {
                resultSet.close();
            }
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
        return sb.toString();
    }
}
